package com.aotain.project.apollo.topo;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.KafkaSpout2;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import com.aotain.common.DomainConfig;
import com.aotain.common.ZkKafkaConfig;
import com.aotain.kafka.KafkaConfigUtil;
import com.aotain.kafka.KafkaProperties;
import com.aotain.project.apollo.bolt.SourceFmtBolt;
import com.aotain.project.apollo.bolt.UrlAbnorBolt;

public class NGramTopo {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		if (args.length != 3){
			System.err.printf("Eg: WebAttackTopo /user/hive/warehouse/tianwang.db/urlabnormodel/  /user/hive/warehouse/tianwang.db/probaresult/");
			System.exit(1);
		} 
		
		TopologyBuilder builder = new TopologyBuilder();
		ZkKafkaConfig z_kConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");
		String filterIPs = z_kConfig.getFilterIPss();
		//suffixname="cgi|tpt|mp4|htc|bif|m4a|so|ipcc|ipd|patch|trt|cfg|torrent|crl|lcr|fcg|do|m3u8|config|cdr|mkv|tsd|crc|flv|wav|manifest|mpq|mfil|swz|dns|msi|pack|ts|ver|ctl|srf|zip|rar|bmp|ico|pic|exe|doc|docx|pdf|xls|xlsx|gz|f4v|gif|jpg|png|css|json|js|txt|eot|id|gzip|cab|7z|vbs|cer|etf|kwp|swf|dat|ini|dll|zl|msu|lua|asa|psf|jpeg|qpyd|blf|ads|bin|xdelta|webp|xml|lrc|data|e|lst|npk|crt|rp|dtd|msp|rmvb|bup|sdo|org|cit|qgi";
		//domainlist="91baiyin.com.cn|amoney.com.cn|aniu.tv|fujixerox.com.cn|gasgoo.com|muyingzhijia.com"
		DomainConfig d_conf = new DomainConfig("/home/storm/config/dbconfig.ini");
		
		builder.setSpout("sourceSpout", new KafkaSpout2(new KafkaConfigUtil(z_kConfig.getZkConnServer(),KafkaProperties.Abnormal_Log_topic,KafkaProperties.zkRoot,KafkaProperties.Log_Abnormal_groupId2 ).getConfig()),6);
		builder.setBolt("SourceFmtBolt", new SourceFmtBolt(filterIPs),6).shuffleGrouping("sourceSpout");
		builder.setBolt("UrlAbnorBolt", new UrlAbnorBolt(args[1],args[2]),6).fieldsGrouping("SourceFmtBolt", new Fields("domainname"));
		Config conf = new Config(); 
		conf.put("mushroomserver", z_kConfig.getMushroomServer());
		conf.put("zkConn", z_kConfig.getZkConnServer());
		conf.put("suffixname", d_conf.getSuffixName());
		conf.put("domainlist", d_conf.getDomainList());
		conf.setNumWorkers(3);
		conf.setNumAckers(0);
		conf.setDebug(false);
	
		if (args.length > 0) {  
			try {
				StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
			} catch (AlreadyAliveException e) {
				e.printStackTrace();
			} catch (InvalidTopologyException e) {
				e.printStackTrace();
			}	catch (AuthorizationException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}else {
			LocalCluster localCluster = new LocalCluster();
			localCluster.submitTopology("NGramTopo", conf, builder.createTopology());
		}
	}

}
